Conversation
Process N messages per consumer process then shut down. When set, the MCP does not respawn exited processes, so the entire daemon exits once all consumers have reached their limit. Closes #31 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 20 minutes and 23 seconds. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (4)
📝 WalkthroughWalkthroughAdds a CLI flag Changes
Sequence Diagram(s)sequenceDiagram
participant CLI
participant Controller
participant MCP as MasterControlProgram
participant Process
participant Broker as MessageBroker
CLI->>Controller: start (--max-messages=N)
Controller->>MCP: start(max_messages=N)
MCP->>Process: spawn(kwargs{..., max_messages:N})
Process->>Broker: consume message
Broker-->>Process: deliver message
Process->>Process: increment _processed_count
alt _processed_count < max_messages
Process->>Broker: ack / continue
else _processed_count >= max_messages
Process->>Process: log limit reached
Process->>Process: shutdown_connections()
Process->>MCP: exit/terminate
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
rejected/controller.py (1)
166-174: Consider validatingmax_messagesto reject non-positive values.The argument accepts any integer, including 0 or negative values. A value of 0 would be falsy and act as "no limit" (potentially confusing), while negative values would cause immediate shutdown after the first message since
_processed_count >= self.max_messageswould always be true.Add validation in main()
if args.quantity is not None and args.consumer is None: parser.error('--qty requires --only') + if args.max_messages is not None and args.max_messages < 1: + parser.error('--max-messages must be at least 1')🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/controller.py` around lines 166 - 174, The CLI currently allows zero or negative values for the --max-messages argument; add validation in main() (or via a custom argparse type) to reject non-positive integers for max_messages: ensure parser.add_argument for '--max-messages' either uses a custom validator like positive_int or, after parsing, checks args.max_messages and exits/raises an error if args.max_messages is not None and args.max_messages <= 0; reference the parser.add_argument call and the main() function and the max_messages variable so the check prevents 0/negative values that would confuse the _processed_count logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@rejected/controller.py`:
- Around line 166-174: The CLI currently allows zero or negative values for the
--max-messages argument; add validation in main() (or via a custom argparse
type) to reject non-positive integers for max_messages: ensure
parser.add_argument for '--max-messages' either uses a custom validator like
positive_int or, after parsing, checks args.max_messages and exits/raises an
error if args.max_messages is not None and args.max_messages <= 0; reference the
parser.add_argument call and the main() function and the max_messages variable
so the check prevents 0/negative values that would confuse the _processed_count
logic.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 9cef54fb-507f-44eb-bec8-49ee08aeba29
📒 Files selected for processing (4)
rejected/controller.pyrejected/mcp.pyrejected/process.pytests/test_controller.py
Add RejectedException and ProcessingContext to rejected.__init__ so consumers can use `rejected.Consumer`, `rejected.ProcessingContext`, etc. directly. Update README examples to use top-level imports, remove stale Python version badge, update docs URL to GH Pages. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
All code examples now use `import rejected` and `rejected.Consumer`, `rejected.TransactionConsumer`, `rejected.ProcessingContext`, etc. instead of sub-module imports. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The AMQP property is `type`, and pika uses `properties.type`. Rename the Pydantic field to match, eliminating the awkward `message_type` indirection. Consumer.message_type property is unchanged for backward compatibility. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
docs/migrating.md (1)
319-332:⚠️ Potential issue | 🟡 MinorAdd
TransactionConsumer→FunctionalConsumerto the removed APIs table.The table documents removed/replaced APIs but doesn't mention the
TransactionConsumerrename. Users searching this table for migration guidance won't find it.📝 Suggested addition
| 3.x API | 4.0 Replacement | |---------|----------------| | `consumer.SmartConsumer` | `rejected.Consumer` (auto-deserializes) | | `consumer.PublishingConsumer` | `rejected.Consumer` (always could publish) | +| `consumer.TransactionConsumer` | `rejected.FunctionalConsumer` | | `self.io_loop` | `asyncio.get_event_loop()` |🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/migrating.md` around lines 319 - 332, The removed APIs table is missing the rename for TransactionConsumer; add a table row mapping `TransactionConsumer` to `FunctionalConsumer` in the same table where other mappings like `consumer.SmartConsumer` → `rejected.Consumer` appear so users can find migration guidance for TransactionConsumer; ensure the new row follows the existing pipe-delimited format and placement (near other consumer/type renames) and uses backticks for the symbols `TransactionConsumer` and `FunctionalConsumer`.rejected/consumer.py (1)
726-758:⚠️ Potential issue | 🟡 MinorAdd a deprecation alias for
TransactionConsumerto maintain backward compatibility.The rename from
TransactionConsumertoFunctionalConsumeris a breaking change for users importingTransactionConsumer. Since the module already maintains backward compatibility with exception re-exports, add an alias before the exception section:# Deprecated alias for backward compatibility TransactionConsumer = FunctionalConsumerAdditionally, rename the internal method
_process_transactional(line 753) to_process_concurrentor similar to align with the new class name.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/consumer.py` around lines 726 - 758, Add a deprecated alias so existing imports of TransactionConsumer keep working by assigning TransactionConsumer = FunctionalConsumer (place this alias before the exception re-exports as requested), and rename the internal method _process_transactional to _process_concurrent in the FunctionalConsumer class; update all internal references (e.g., the lambda passed in _run_consumer that currently calls self._process_transactional) to call self._process_concurrent to keep behavior intact.
🧹 Nitpick comments (1)
docs/migrating.md (1)
352-368: AddTransactionConsumer→FunctionalConsumerto the migration checklist.The checklist is comprehensive but should include the class rename for users of the concurrent consumer.
📝 Suggested addition
6. Replace `SmartConsumer` / `PublishingConsumer` with `rejected.Consumer` +6a. Replace `TransactionConsumer` with `rejected.FunctionalConsumer` 7. Replace `tornado` imports with `asyncio` equivalents🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/migrating.md` around lines 352 - 368, Add an entry to the Quick Migration Checklist noting the class rename from TransactionConsumer to FunctionalConsumer so users of the concurrent consumer update references; specifically mention replacing any uses/imports of TransactionConsumer with FunctionalConsumer (and update docs/tests/configs accordingly), and place this item near other consumer-related entries such as the existing notes about SmartConsumer/PublishingConsumer → rejected.Consumer to keep related migration steps together.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/migrating.md`:
- Around line 100-116: Update the migration note to state that
FunctionalConsumer is a rename of TransactionConsumer rather than a brand-new
class: change the heading and text to indicate "FunctionalConsumer (renamed from
TransactionConsumer)" and replace "4.0 introduces FunctionalConsumer" with
wording like "FunctionalConsumer replaces/renames TransactionConsumer from 3.x"
and add a short note instructing users that code referencing TransactionConsumer
should be updated to use FunctionalConsumer (keeping reference to
rejected.FunctionalConsumer and TransactionConsumer so readers can locate the
symbols).
In `@rejected/__init__.py`:
- Around line 34-40: The top-level export list removed TransactionConsumer which
will break users importing rejected.TransactionConsumer; restore compatibility
by re-introducing a top-level alias/export named TransactionConsumer that
references the original implementation (e.g., the class or function in its
defining module) and add a deprecation notice in its docstring or runtime
warning to guide users to the new import path, then run tests to ensure imports
still resolve (referencing symbol TransactionConsumer and __init__.py to locate
where to add the alias).
In `@rejected/models.py`:
- Line 145: The public Message constructor removed the message_type input which
breaks external callers; restore a compatibility shim by accepting a
message_type parameter in the Message initializer (or model signature) and
mapping it to the new type field when present (e.g., prefer message_type if
provided, otherwise use type), and emit a DeprecationWarning to signal it's
deprecated; update the Message class/constructor (and any factory functions) to
accept message_type: str | None and set self.type accordingly so external code
continues to work.
---
Outside diff comments:
In `@docs/migrating.md`:
- Around line 319-332: The removed APIs table is missing the rename for
TransactionConsumer; add a table row mapping `TransactionConsumer` to
`FunctionalConsumer` in the same table where other mappings like
`consumer.SmartConsumer` → `rejected.Consumer` appear so users can find
migration guidance for TransactionConsumer; ensure the new row follows the
existing pipe-delimited format and placement (near other consumer/type renames)
and uses backticks for the symbols `TransactionConsumer` and
`FunctionalConsumer`.
In `@rejected/consumer.py`:
- Around line 726-758: Add a deprecated alias so existing imports of
TransactionConsumer keep working by assigning TransactionConsumer =
FunctionalConsumer (place this alias before the exception re-exports as
requested), and rename the internal method _process_transactional to
_process_concurrent in the FunctionalConsumer class; update all internal
references (e.g., the lambda passed in _run_consumer that currently calls
self._process_transactional) to call self._process_concurrent to keep behavior
intact.
---
Nitpick comments:
In `@docs/migrating.md`:
- Around line 352-368: Add an entry to the Quick Migration Checklist noting the
class rename from TransactionConsumer to FunctionalConsumer so users of the
concurrent consumer update references; specifically mention replacing any
uses/imports of TransactionConsumer with FunctionalConsumer (and update
docs/tests/configs accordingly), and place this item near other consumer-related
entries such as the existing notes about SmartConsumer/PublishingConsumer →
rejected.Consumer to keep related migration steps together.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 89d8205b-a692-424d-b907-ea3c69605cc9
📒 Files selected for processing (14)
CLAUDE.mdREADME.mddocs/api.mddocs/configuration.mddocs/consumer_howto.mddocs/index.mddocs/migrating.mdrejected/__init__.pyrejected/consumer.pyrejected/models.pyrejected/process.pyrejected/testing.pytests/test_consumer.pytests/test_process.py
✅ Files skipped from review due to trivial changes (3)
- docs/index.md
- docs/configuration.md
- CLAUDE.md
🚧 Files skipped from review as they are similar to previous changes (1)
- rejected/process.py
- Update all code examples in index.md and consumer_howto.md to use `import rejected` / `rejected.Consumer` style - Update pyproject.toml docs URL from readthedocs to GH Pages Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Add a
-n N/--max-messages NCLI flag that causes each consumer process to shut down after processing N messages. When set, the MCP skips respawning exited processes, so the entire daemon exits once all consumers hit their limit.Useful for debugging consumers against real queues without leaving them running indefinitely.
Usage:
rejected -c config.yaml -o my_consumer -n 5Closes #31
Changes
controller.py: Add-n/--max-messagesarg, pass through to MCPmcp.py: Acceptmax_messages, pass to Process kwargs, skipcheck_process_countswhen setprocess.py: Addmax_messagesproperty, check limit inon_processedand initiate graceful shutdowntests/test_controller.py: Addmax_messagesto mock args fixtureTest plan
rejected -c example.yaml -o consumer -n 3processes 3 messages then exits🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation
type.Exports
Tests